<?php

namespace App\Core\Event;

use App\Core\Process\Daemon;
use App\Entities\HllEvent;
use App\Events\EventHanderUtils;
use App\Exceptions\ProcessUnrecoverableException;
use App\Loggers\MallLogger;
use App\Services\AppErrorService;
use App\Services\Dev\HllEventService;
use App\Utils\AmqpConnection;
use Illuminate\Console\OutputStyle;
use PhpAmqpLib\Message\AMQPMessage;
use Predis\ClientException;

class RabbitMqConsumer
{

    use Daemon;

    /**
     * The output interface implementation.
     *
     * @var
     */
    protected $output;

    /**
     * @param \Illuminate\Console\OutputStyle $output
     */
    public function setOutput($output): void
    {
        $this->output = $output;
    }



    public function consume($queueName)
    {
        $logger = app('logger');
        $this->logger = $logger;
        $this->signalRegistry();
        /** @var AmqpConnection $amqp */
        $amqp = app(AmqpConnection::class);
        $conn = $amqp->connect();
        $channel = $conn->channel();
        EventHanderUtils::checkAction($queueName);
        $exchange = EventHanderUtils::getExchange($queueName);
        $routekeys = EventHanderUtils::getRouteKey($queueName);
        $channel->queue_declare($queueName, false, true, false, false);
        foreach ($routekeys as $routekey) {
            $channel->queue_bind($queueName, $exchange, $routekey);
        }
        $hllEventService = new HllEventService();
        $hllEventService->setEm(app('em'));
        //将 类型消息 绑定到该队列
        /** @var AMQPMessage $msg */
        $callback = function ($msg) use ($logger, $queueName, $hllEventService) {
            $msg->ack(); //防止因错误导致的，无限requeue。这种错误的消息，交给panel手工重试
            $body = json_decode($msg->body, true);
            $handlers = EventHanderUtils::getHandler($queueName);
            /** @var AbstractMessage $message */
            $message = objectMapper($body['data'], AbstractMessage::class);
            $logger->setTraceNo($queueName . '_' . $message->getBizId());
            $hllevent = $hllEventService->findEvent($message->getId());
            $updateEvent = new HllEvent();
            try {
                $errorTasks = [];
                if (isset($hllevent->error_tasks) && $hllevent->error_tasks) {
                    $errorTasks = explode(',', $hllevent->error_tasks);
                    $this->info("处理重试消息");
                }
                $updateEvent->setEstatus(HllEventService::ESTATUS_DOING);
                $hllEventService->updateEventByMessageId($message->getId(), $updateEvent);
                $taskCount = count($handlers);
                $failNums = 0;
                $errorClass = [];
                $errorInfo = [];
                foreach ($handlers as $handler) {
                    $handlerClass = get_class($handler);
                    if ($errorTasks && !in_array($handlerClass, $errorTasks)) {
                        continue;
                    }
                    $logger->setTraceNo($handlerClass . '_' . $message->getBizId());
                    try {
                        $this->info("{$handlerClass} 处理数据 " . $body['action'] . ' ' . $body['data']);
                        $handler->handle($body);
                    } catch (\Throwable $exc) {
                        $failNums++;
                        $errorClass[] = $handlerClass;
                        $errorInfo[] = [
                            $handlerClass => $exc->getMessage()
                        ];
                        $this->error("执行 {$handlerClass} 失败" . $exc->getMessage());
                        $this->subtasksExceptionHand($exc);
                        continue;
                    }
                }
                $finishStatus = $this->completionStatus($taskCount, $failNums);
                $updateEvent->setEstatus($finishStatus);
                $updateEvent->setErrorTasks(implode(',', $errorClass));
                $updateEvent->setErrorInfo(json_encode($errorInfo));
                $hllEventService->updateEventByMessageId($message->getId(), $updateEvent);
            } catch (\Throwable $exc) {
                $updateEvent->setEstatus(HllEventService::ESTATUS_FAIL);
                $updateEvent->setErrorTasks('');
                $updateEvent->setErrorInfo($exc->getMessage());
                $hllEventService->updateEventByMessageId($message->getId(), $updateEvent);
                $this->error($exc->getMessage());
                $logger->error($exc->getMessage(), AppErrorService::formatExceptionObject($exc));
                if ($exc instanceof ProcessUnrecoverableException) {
                    throw $exc;
                }
            }
        };
        // 消费 队列
        $channel->basic_consume($queueName, '', false, false, false,
            false, $callback);
        // 阻塞队列监听事件
        $num = 0;
        while ($channel->is_open() && !$this->processExit) {
            $num++;
            $channel->wait();
            if (($num % 100) == 0) {
                $this->info("手动gc");
                gc_collect_cycles();
            }
        }
    }

    public function info($string, $verbosity = null)
    {
        $mem = (memory_get_usage() / 1024 / 1024) . 'MB';
        $string .= " memory_usage $mem";
        if($this->output instanceof OutputStyle){
            $pid = posix_getpid();
            $this->output->info("[$pid] ".$string);
        }
        app('logger')->info($string);
    }

    /**
     * @param $taskCount
     * @param $failNums
     * @return int
     */
    private function completionStatus($taskCount, $failNums)
    {
        if ($failNums == 0) {
            return HllEventService::ESTATUS_DONE;
        }
        if ($failNums == $taskCount) {
            return HllEventService::ESTATUS_FAIL;
        }
        return HllEventService::ESTATUS_PARTIAL_SUCCESS;
    }

    public function error($message)
    {
        if($this->output instanceof OutputStyle){
            $pid = posix_getpid();
            $this->output->error("[{$pid}] ".$message);
        }

    }

    protected function subtasksExceptionHand(\Throwable $exc)
    {
        /** @var MallLogger $logger */
        $logger = app('logger');
        $logger->error('MQ消费异常', AppErrorService::formatExceptionObject($exc));
        if ($exc instanceof ClientException) {
            throw new ProcessUnrecoverableException($exc->getMessage(), 2500, $exc);
        }
    }

}